package com.ricky.mq.client.code.listener;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.IdUtil;
import com.ricky.mq.client.code.constant.ClientCommandCode;
import com.ricky.mq.client.code.handler.WaitOrSendMsgHandler;
import com.ricky.mq.client.message.BoringMessage;
import com.ricky.mq.client.message.data.client.ClientMessageListenerData;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 消息分派器
 *
 * @author ricky
 * @since 2020.06.18
 */
public class MsgAssignment {

    /**
     * 每个queue与它自身的监听器
     */
    private static Map<String, List<MessageListener>> queueListenerMap = new ConcurrentHashMap<>(16);
    /**
     * 每个topic与它自身的监听器
     */
    private static Map<String, List<MessageListener>> topicListenerMap = new ConcurrentHashMap<>(16);

    /**
     * 添加一个队列监听器
     *
     * @param queueName
     * @param messageListener
     */
    public static void addQueueMessageListener(String queueName, MessageListener messageListener) {

        WaitOrSendMsgHandler.asyncSendData(ClientCommandCode.LISTENER_QUEUE_MESSAGE, ClientMessageListenerData.build().setMsgModelName(queueName).setUuid(IdUtil.fastSimpleUUID()));

        if (CollUtil.isEmpty(queueListenerMap.get(queueName))) {
            synchronized (queueListenerMap) {
                if (CollUtil.isEmpty(queueListenerMap.get(queueName))) {
                    List<MessageListener> messageListeners = new ArrayList<>();
                    queueListenerMap.put(queueName, messageListeners);
                }
            }
        }
        queueListenerMap.get(queueName).add(messageListener);
    }

    /**
     * 添加一个广播监听器
     *
     * @param topicName
     * @param messageListener
     */
    public static void addTopicMessageListener(String topicName, MessageListener messageListener) {

        WaitOrSendMsgHandler.asyncSendData(ClientCommandCode.LISTENER_TOPIC_MESSAGE, ClientMessageListenerData.build().setMsgModelName(topicName).setUuid(IdUtil.fastSimpleUUID()));

        if (CollUtil.isEmpty(topicListenerMap.get(topicName))) {
            synchronized (topicListenerMap) {
                if (CollUtil.isEmpty(topicListenerMap.get(topicName))) {
                    List<MessageListener> messageListeners = new ArrayList<>();
                    topicListenerMap.put(topicName, messageListeners);
                }
            }
        }
        topicListenerMap.get(topicName).add(messageListener);
    }

    /**
     * 派发queue消息
     * @param queueName
     * @param boringMessage
     */
    public static void assignmentQueueMsg(String queueName, BoringMessage boringMessage) {
        queueListenerMap.get(queueName).forEach((messageListener) -> {
            messageListener.listenerMessage(boringMessage);
        });
    }

    /**
     * 派发topic消息
     * @param topicName
     * @param boringMessage
     */
    public static void assignmentTopicMsg(String topicName, BoringMessage boringMessage) {
        topicListenerMap.get(topicName).forEach((messageListener) -> {
            messageListener.listenerMessage(boringMessage);
        });
    }

}
